#include "threadpool.h"

struct threadpool *create_threadpool(int threads, int queue_length, void *(*con)(void *))
{
    struct threadpool *pool = malloc(sizeof(struct threadpool));
    int ret;
    pool->admin_id = malloc(1024*sizeof(long));
    pool->equipment_id = malloc(1024*sizeof(long));
    pool->sensor_id = malloc(1024*sizeof(long));
    memset(pool->admin_id, 0xFF, 1024*sizeof(long));
    memset(pool->equipment_id, 0xFF, 1024*sizeof(long));
    memset(pool->sensor_id, 0xFF, 1024*sizeof(long));
    pool->tids = malloc(sizeof(pthread_t) * threads);
    pool->que = create_queue();
    pool->threads = threads; 
    pool->conn = getnewconn();
    pool->numofcit=0;
    pool->queue_length_max = queue_length;
    sem_init(&pool->semw, 0, queue_length);
    sem_init(&pool->semr, 0, 0);
    pthread_mutex_init(&pool->mutex, NULL);
    pool->runflag = 1;
    for (int i = 0; i < threads; i++)
    {
        ret = pthread_create(&pool->tids[i], NULL, con, pool);
        if (ret != 0)
        {
            perror("pthread_create");
            return NULL;
        }
    }

    return pool;
}

void destroy_threadpool(struct threadpool *pool)
{
    int ret;
    for (int i = 0; i < pool->threads; i++) // 使用存储的线程数量
    {
        if ((ret = sem_post(&pool->semr)) != 0)
        {
            perror("sem_post");
            fprintf(stderr, "Error code: %d\n", ret);
        }
    }
    if ((ret = sem_post(&pool->semw)) != 0)
    {
        perror("sem_post");
        fprintf(stderr, "Error code: %d\n", ret);
    }

    for (int i = 0; i < pool->threads; i++)
    {
        if ((ret = pthread_join(pool->tids[i], NULL)) != 0)
        {
            perror("pthread_join");
            fprintf(stderr, "Error code: %d\n", ret);
        }
    }
    free(pool->tids);
    free(pool->admin_id);
    free(pool->equipment_id);
    free(pool->sensor_id);
    destroy_queue(pool->que);
    sem_destroy(&pool->semr);
    sem_destroy(&pool->semw);
    pthread_mutex_destroy(&pool->mutex);
    mysql_close(pool->conn);
    free(pool);
}
